package com.cskaoyan.order.mq.delay;

import com.cskaoyan.order.dal.entitys.Order;
import com.cskaoyan.order.dal.entitys.OrderItem;
import com.cskaoyan.order.dal.entitys.Stock;
import com.cskaoyan.order.dal.persistence.OrderItemMapper;
import com.cskaoyan.order.dal.persistence.OrderMapper;
import com.cskaoyan.order.dal.persistence.StockMapper;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.util.Date;
import java.util.List;

/**
 * @auther cskaoyan
 * @date 2022/6/21:17:32
 */
@Component
public class DelayOrderCancelConsumer {

    private DefaultMQPushConsumer consumer;

    @Autowired
    OrderMapper orderMapper;
    @Autowired
    StockMapper stockMapper;
    @Autowired
    OrderItemMapper orderItemMapper;

    @PostConstruct
    public void init() {
        consumer = new DefaultMQPushConsumer("delay_order_cancel_consumer");

        consumer.setNamesrvAddr("127.0.0.1:9876");

        try {
            // 订阅主题
            consumer.subscribe("delay_order_cancel", "*");

            consumer.setMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    // 1. 从message中获取orderId
                    // 2. 根据orderId查询，订单状态，如果发现，订单已支付，或已取消，什么都不做
                    // 3. 否则，如果发现，订单的状态仍然是初始化，此时，就可以取消该订单
                    //    a. 修改订单状态为已取消
                    //    b. 根据orderId，查询所有订单商品条目，根据每一个订单条目购买数量  -lock_count  +stock_count
                    MessageExt message = msgs.get(0);
                    byte[] body = message.getBody();
                    try {
                        String orderId = new String(body, 0, body.length, "utf-8");
                        Order order = orderMapper.selectByPrimaryKey(orderId);
                        if (order.getStatus() == 0) {
                            orderMapper.updateStatusAndUpdateTime(7, new Date(), orderId);
                            List<OrderItem> orderItems = orderItemMapper.queryByOrderId(orderId);
                            for (OrderItem orderItem : orderItems) {
                                Stock stock = new Stock();
                                stock.setItemId(orderItem.getItemId());
                                stock.setLockCount(-orderItem.getNum());
                                stock.setStockCount(((long) orderItem.getNum()));
                                stockMapper.updateStock(stock);
                            }
                        }
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 启动消费者
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }


}
